package com.atguigu.transform;

import com.atguigu.func.ClickSource;
import com.atguigu.pojo.Event;
import com.atguigu.pojo.Wordcount;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author Mr.Zheng
 * @Date 2023/6/13 22:24
 *
 * reduce:规约聚合
 *      两两聚合,同一个key的第一条数据到来，直接输出，不进reduce方法
 *      从第二条数据往后，每来一条数据都会进入reduce方法进行两两合并
 *
 */
public class Flink03_ReduceTransform {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度
        env.setParallelism(1);
        //获取数据
        DataStreamSource<Event> ds = env.addSource(new ClickSource());

        ds.print("input");

        //统计每个user的点击次数 ， 通过reduce来实现。
        ds.map(
                event -> new Wordcount(event.getUser() , 1 )
        ).keyBy(
                Wordcount::getWord
        ).reduce(
                new ReduceFunction<Wordcount>() {
                    /*
                       两个参数：
                          value1:  要合并的第一个值
                          value2:  要合并的第二个值
                     */
                    @Override
                    public Wordcount reduce(Wordcount value1, Wordcount value2) throws Exception {
                        System.out.println("reduce.....");
                        return new Wordcount(value1.getWord()  , value1.getCount() + value2.getCount());
                    }
                }
        );
        //.print("reduce");

        //需求：统计所有用户中访问频次最高的
        //首先统计每个用户的访问次数
        ds.map(
                event -> new Wordcount(event.getUser(),1)
        ).keyBy(
                Wordcount::getWord
        ).reduce(
                (value1,value2) -> new Wordcount(value1.getWord(),value1.getCount() + value2.getCount())
        ).keyBy(
                wordcount -> true
        ).reduce(
                (value1,value2) ->{
                    if(value1.getCount() > value2.getCount()){
                        return value1;
                    }
                    else
                        return value2;
                }
        ).print("max");


        env.execute();


    }
}
